/*!
巧妙运用broadcast，channel完成优雅关机问题

 - broadcast (Sender关闭，rx.recv将收到Closed)，所有线程监听此rx，收到信息则退出
 - channel(Sender关闭，rx.recv收到None)，所有线程都分配一个tx, 主线程监听rx, 如果所有工作线程都退出，则tx关闭，则主线程rx收到信息（None），表示完成收尾工作。

 当前案例分别启动了Deamon和Listener两个主服务，所以针对 notify_shutdown, shutdown_complete_tx 分别做了clone,
 因此，需要做2此显式的drop，才能正确的关闭这2中通信管道。drop(notify_shutdown); drop(shutdown_complete_tx);

 */

use crate::shutdown::Shutdown;
use crate::worker::*;
use crate::web::*;
use crate::*;

use crate::Cli;
use std::sync::Arc;
use tokio::net::TcpListener;
use tokio::sync::{broadcast, mpsc};
use log::Level::{Error, Info};

#[cfg(feature = "rotatelog")]
use log::{debug, error, info, warn, log as event};
#[cfg(feature = "tracelog")]
use tracing::{error, info, event};


#[derive(Debug)]
struct Deamon {
    notify_shutdown: broadcast::Sender<()>,
    shutdown_complete_rx: mpsc::Receiver<()>,
    shutdown_complete_tx: mpsc::Sender<()>,
}

pub async fn startup(cli: Cli) -> Result<()> {
    let (notify_shutdown, _) = broadcast::channel(1);
    let (shutdown_complete_tx, shutdown_complete_rx) = mpsc::channel(1);

    let mut server = Deamon {
        notify_shutdown,
        shutdown_complete_tx,
        shutdown_complete_rx,
    };
    server.run(cli).await.unwrap();

    // 前面不能阻塞！
    info!("注意：看见这里，各线程才能安全退出！");
    tokio::select! {
        _ = shutdown_signal("main") => {
            error!("try to signal shutting down all servers now!");
        },
    }

    let Deamon {
        mut shutdown_complete_rx,
        shutdown_complete_tx,
        notify_shutdown,
        ..
    } = server;

    drop(notify_shutdown);
    drop(shutdown_complete_tx);

    /*
    这里注意，因为前面对2种管道进行了clone， 所以必须相应 drop

    shutdown_complete_tx 放入每个work线程中，作用就是表示完成收尾工作，安全退出了
    只要有一个shutdown_complete_tx引用存在，下面的recv就会阻塞，直到工作线程退出（释放给他的shutdown_complete_tx）
    */
    let _ = shutdown_complete_rx.recv().await;
    info!("各工作线程安全关闭！");
    Ok(())
}

impl Deamon {
    async fn run(&mut self, cli: Cli) -> crate::Result<()> {
        info!("accepting inbound connections");
        let sync_timer = if cli.mq.len() == 0 {
            DEFAULT_FAST_SYNC_TIMER
        } else {
            DEFAULT_SLOW_SYNC_TIMER
        };

        let workpath = if cli.data.len() > 0 {
            info!("set work path =  {:#?}", &cli.data);
            cli.data.clone()
        } else {
            let program = env::args().next().unwrap();
            let mut wpath = PathBuf::new();
            wpath.push(program);
            let ppath = wpath.as_path().parent().unwrap();
            let root = format!("root{}", cli.node_id);
            ppath.join(root).to_string_lossy().to_string()
        };

        let wroot = PathBuf::from(workpath.as_str());
        std::fs::create_dir_all(wroot).expect("创建根路径失败！");
        // 创建Gfs对象
        let gfs = Gfs::new(cli.master.clone(), workpath.as_str()).await;
        // 创建基于tantivy index 的搜索引擎NodeIndex对象
        let index = NodeIndex::new(
            PathBuf::from(workpath.as_str()),
            DEFUALT_INDEX_NAME,
            DEFAULT_INDEX_MSIZE,
        )?;

        let channelclcient = Arc::new(rabbitmq::client::ChannelClient::new());
        let shared = Arc::new(ShareComm::new(
            gfs,
            index,
            cli.mq.as_str(),
            cli.token.as_str(),
            cli.port,
            Arc::clone(&channelclcient),
        ));
       
        //
        // rabbitmq deamon         
        // 如果设置了mq地址，并且不是redis, 且是master，则启动内置MQ服务
        //
        if cli.mq.len() > 0 && !cli.mq.starts_with("redis") && shared.is_master() {
            let mqserver_str = Box::leak(format!("{}", &cli.mq).into_boxed_str());
            let token = Box::leak(format!("{}", &cli.token).into_boxed_str());
            let cclient = Arc::clone(&channelclcient);
            let _ = tokio::spawn(async move {
                let r = rabbitmq_main(mqserver_str, token, cclient).await;
                if let Err(err) = r {
                    error!("{}, {}", err, "启动内置MQ Server失败！");
                }
            });
            event!(target:"startup", loglevel(Info), "Rabbitmq(inner-MQ server) is startup, OK");
        }

        let cclient = Arc::clone(&channelclcient);
        // 
        // worker deamon
        // 执行数据同步的任务，新文件的内容提取，索引创建等。
        // 
        let mut worker = Worker {
            rmq_client: cclient,
            shared: Arc::clone(&shared),
            sync_timer,
            // Receive shutdown notifications.
            shutdown: Shutdown::new(self.notify_shutdown.subscribe()),
            _shutdown_complete: self.shutdown_complete_tx.clone(),
        };
        let _ = tokio::spawn(async move {
            if let Err(err) = worker.run().await {
                error!("{} {}", err, "connection error");
            }
        });

        //
        // parallel
        // 并行计算服务线程，用于接收并行计算的任务
        //
        let mut parallel = Parallel {
            shared: Arc::clone(&shared),
            // Receive shutdown notifications.
            shutdown: Shutdown::new(self.notify_shutdown.subscribe()),
            _shutdown_complete: self.shutdown_complete_tx.clone(),
        };

        let _ = tokio::spawn(async move {
            if let Err(err) = parallel.run().await {
                error!("connection error: {}", err);
            }
        });

        //
        // web server 
        // 启动web服务线程
        //
        let wshared = Arc::clone(&shared);
        let _ = tokio::spawn(async move {
            if let Err(err) = axum_main(wshared, cli.port).await {
                error!("{:#?}, {}", err, "axum server error");
            }else{
                event!(target:"shutdown", loglevel(Error), "Web server is shutdown, OK");
            }
        });
        event!(target:"startup", loglevel(Info), "Web server(based on axum) is startup, OK");

        //
        //  启动MQ consumer
        //  如果提供了MQ地址，并且不是Master，则会启动MQ Consumer，用于接收同步数据的指令。
        //
        if cli.mq.len() > 0 && !shared.is_master() {
            let group = Box::leak(format!("gfs-node-{}", cli.node_id).into_boxed_str());
            let consumer_name = Box::leak(
                format!("gfs-node-{}-slave-{}", cli.node_id, cli.slave_id).into_boxed_str(),
            );
            let cclient = Arc::clone(&channelclcient);
            let mshared = Arc::clone(&shared);

            let mut consumer = MqConsumer{   
                rmq_client: cclient,
                shutdown:  Shutdown::new(self.notify_shutdown.subscribe()),
                _shutdown_complete: self.shutdown_complete_tx.clone(),
                shared: mshared,
            };

            let _ = tokio::spawn(async move {
                if let Err(err) = consumer.run(group, consumer_name).await {
                    error!("{}, {}", err, "axum server error");
                }
            });           

        } else {
            event!(target:"startup", loglevel(Info), "HERE IS MASTER OR SLAVE WHITOUT MQ, SO NO MQ CONSUMER !");
        }
        info!("finished Demaon runner, see here is ok!");
        Ok(())
    }
}


pub async fn rabbitmq_main(
    mq_server: &str,
    token: &str,
    channelclient: Arc<rabbitmq::client::ChannelClient>,
) -> Result<()> {
    // Bind a TCP listener
    let port = if mq_server.find(":") != None {
        let sv: Vec<&str> = mq_server.split(":").collect();
        sv[sv.len() - 1]
    } else {
        mq_server
    };
    let listener = TcpListener::bind(&format!("127.0.0.1:{}", port)).await?;
    info!("inner rabbitmq is run at 127.0.0.1:{}", port);
    rmqserver::run(listener, token, channelclient, shutdown_signal("rabbitmq") ).await;
    event!(target:"shutdown", loglevel(Error), "Rabbitmq(inner-MQ server) is shutdown, OK");
    Ok(())
}

///////

